package global

import (
	"fmt"
	"github.com/streadway/amqp"
	"log"
	"longmen/server/pkg/common/helper/const"
	"sync"
)

/**
 * @Author admin
 * @Description 每一个APP 仅创建一个CONN, 每次使用之前创建 Channel ,如果不是业务需要的长连接，请手动关闭channel.
 * @Date 9:44 2022/10/4
 * @Param
 * @return
 **/
var giftRabbitMQ *RabbitMQ
var lock sync.Mutex

func GetGiftRabbitMQ() *RabbitMQ {
	return giftRabbitMQ
}

/**
 * @Author admin
 * @Description //所有的队列初始化在这里
 * @Date 9:49 2022/10/4
 * @Param
 * @return
 **/
func InitMQ() {
	mqUrl := Config.GetString("amqp.url")
	conn := NewConn(mqUrl)
	initGiftMQ(conn, mqUrl)
}
func initGiftMQ(conn *amqp.Connection, mqUrl string) {

	//仅仅是建创绑定
	giftRabbitMQ = NewRabbitMQRouting(_const.MQ_EXCHANGE_LIVE_GIFT, _const.MQ_QUEUE_LIVE_GIFT, _const.MQ_ROUTE_KEY_LIVE_GIFT, mqUrl, conn)
	channel := giftRabbitMQ.BindExchangeDeclare()
	giftRabbitMQ.CloseChannel(channel)
}

/**
 * @Author admin
 * @Description声明一个rabbit结构体
 * @Date 10:52 2022/10/1
 * @Param
 * @return
 **/
type RabbitMQ struct {
	conn *amqp.Connection
	//channel *amqp.Channel
	//队列名称
	QueueName string
	//交换机名称
	Exchange string
	//bind Key 名称
	Key string
	//连接信息
	MqUrl string
}

/**
 * @Author admin
 * @Description创建结构体实例
 * @Date 10:52 2022/10/1
 * @Param
 * @return
 **/
func NewRabbitMQ(queueName, exchange, key, mqUrl string, conn *amqp.Connection) *RabbitMQ {
	return &RabbitMQ{QueueName: queueName, Exchange: exchange, Key: key, MqUrl: mqUrl, conn: conn}
}

/**
 * @Author admin
 * @Description 断开channel 和 connection
 * @Date 10:52 2022/10/1
 * @Param
 * @return
 **/
func (r *RabbitMQ) Destory(channel *amqp.Channel) {
	channel.Close()
	r.conn.Close()
}

/**
 * @Author admin
 * @Description //TODO
 * @Date 9:27 2022/10/4
 * @Param
 * @return
 **/
func (r *RabbitMQ) GetConn() *amqp.Connection {
	return r.conn
}

/**
 * @Author admin
 * @Description 错误处理函数
 * @Date 10:53 2022/10/1
 * @Param
 * @return
 **/
func (r *RabbitMQ) failOnErr(err error, message string) {
	if err != nil {
		panic(fmt.Sprintf("%s:%s", message, err))
	}
}

/**
 * @Author admin
 * @Description 创建RabbitMQ实例
 * @Date 10:54 2022/10/1
 * @Param
 * @return
 **/
func NewRabbitMQRouting(exchangeName, queueName, routingKey, mqUrl string, conn *amqp.Connection) *RabbitMQ {
	//创建RabbitMQ实例
	rabbitmq := NewRabbitMQ(queueName, exchangeName, routingKey, mqUrl, conn)
	//rabbitmq.NewConn(rabbitmq.MqUrl)
	return rabbitmq
}

func NewConn(mqUrl string) *amqp.Connection {
	defer func() {
		if err := recover(); err != nil {
			NewConn(mqUrl)
		}
	}()
	var err error
	var conn *amqp.Connection
	//获取connection
	lock.Lock()
	defer lock.Unlock()
	if conn == nil {
		conn, err = amqp.Dial(mqUrl)
		if err != nil {
			log.Fatalf("%s:%s", "failed to connect rabbitmq!", err)
			panic(fmt.Sprintf("%s:%s", "failed to connect rabbitmq!", err))
		}
	}
	return conn
}

func (r *RabbitMQ) ReConn() {
	defer func() {
		if err := recover(); err != nil {
			r.ReConn()
		}
	}()
	var err error
	//获取connection
	lock.Lock()
	defer lock.Unlock()
	if r.conn == nil || r.conn.IsClosed() {
		if r.conn, err = amqp.Dial(r.MqUrl); err != nil {
			r.failOnErr(err, "failed to connect rabbitmq!")
		}
	}
}

func (r *RabbitMQ) OpenChannel() *amqp.Channel {
	//获取channel
	var channel *amqp.Channel
	var err error
	channel, err = r.conn.Channel()
	if err != nil {
		fmt.Println(fmt.Sprintf("%s:%s", "failed to open a channel", err))
		//r.failOnErr(err, "failed to open a channel")
		if r.conn.IsClosed() {
			r.ReConn()
		}
		channel = r.OpenChannel()
	}
	return channel
}

func (r *RabbitMQ) CloseChannel(channel *amqp.Channel) {
	//关闭channel
	defer channel.Close()
}

/**
 * @Author admin
 * @Description //TODO
 * @Date 11:20 2022/10/1
 * @Param
 * @return
 **/
func (r *RabbitMQ) BindExchangeDeclare() *amqp.Channel {
	channel := r.OpenChannel()
	//1.尝试创建交换机
	err := channel.ExchangeDeclare(r.Exchange, amqp.ExchangeDirect, true, false, false, false, nil)
	r.failOnErr(err, "Failed to declare an exchange")
	//创建队列
	q, err := channel.QueueDeclare(r.QueueName, true, false, false, false, amqp.Table{
		"x-message-ttl":             1800000,                            //半小时,消息过期时间,毫秒 0 立即投递
		"x-dead-letter-exchange":    fmt.Sprintf("dead_%s", r.Exchange), // 指定死信交换机
		"x-dead-letter-routing-key": fmt.Sprintf("dead_%s", r.Key),      // 指定死信routing-key
	})
	r.failOnErr(err, "Failed to declare a queue")
	//绑定队列到 exchange 中
	err = channel.QueueBind(q.Name, r.Key, r.Exchange, false, nil)
	// 声明死信队列
	// args 为 nil。切记不要给死信队列设置消息过期时间,否则失效的消息进入死信队列后会再次过期。
	_, err = channel.QueueDeclare(fmt.Sprintf("dead_%s", r.QueueName), true, false, false, false, nil)
	// 声明交换机
	err = channel.ExchangeDeclare(fmt.Sprintf("dead_%s", r.Exchange), amqp.ExchangeDirect, true, false, false, false, nil)
	// 队列绑定（将队列、routing-key、交换机三者绑定到一起）
	err = channel.QueueBind(fmt.Sprintf("dead_%s", r.QueueName), fmt.Sprintf("dead_%s", r.Key), fmt.Sprintf("dead_%s", r.Exchange), false, nil)

	return channel
}

/**
 * @Author admin
 * @Description 路由模式发送消息
 * @Date 10:55 2022/10/1
 * @Param
 * @return
 **/
func (r *RabbitMQ) PublishRouting(message string) {
	//1.绑定交换机信息
	channel := r.BindExchangeDeclare()
	defer channel.Close()
	//2.发送消息
	err := channel.Publish(
		r.Exchange,
		//要设置
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	if err != nil {
		fmt.Println(err)
	}
}

/**
 * @Author admin
 * @Description 路由模式发送消息
 * @Date 10:55 2022/10/1
 * @Param
 * @return
 **/
func (r *RabbitMQ) PublishRoutingWithOutBind(channel *amqp.Channel, message string) {
	//2.发送消息
	err := channel.Publish(
		r.Exchange,
		//要设置
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	if err != nil {
		fmt.Println(err)
	}
}

/**
 * @Author admin
 * @Description 路由模式发送消息 带确认的方式实例
 * @Date 10:55 2022/10/1
 * @Param
 * @return
 **/
/*func (r *RabbitMQ) PublishConfirmMessage(channel *amqp.Channel, message string) {
	channel.Confirm(false)
	confirms := channel.NotifyPublish(make(chan amqp.Confirmation, 1)) // 处理确认逻辑
	//2.发送消息
	err := channel.Publish(
		r.Exchange,
		//要设置
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	if err != nil {
		fmt.Println(err)
	}
	if confirmed := <-confirms; confirmed.Ack {
		fmt.Printf("confirmed delivery with delivery tag: %d", confirmed.DeliveryTag)
	} else {
		fmt.Printf("confirmed delivery of delivery tag: %d", confirmed.DeliveryTag)
	}
}*/

// 路由模式接受消息
func (r *RabbitMQ) RecieveRouting(autoAck bool) (<-chan amqp.Delivery, error) {
	channel := r.BindExchangeDeclare()
	//消费消息
	return channel.Consume(
		r.QueueName,
		"",
		autoAck,
		false,
		false,
		false,
		nil,
	)
}

func (r *RabbitMQ) RecieveRoutingWithOutBind(channel *amqp.Channel) (<-chan amqp.Delivery, error) {
	//defer channel.Close()
	//消费消息
	return channel.Consume(
		r.QueueName,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
}

/*
func main() {
	go test()

	channel := giftRabbitMQ.OpenChannel()
	defer channel.Close()
	for i := 0; i <= 10; i++ {
		channel.Tx()
		giftRabbitMQ.PublishRoutingWithOutBind(channel, "Hello kuteng one!"+strconv.Itoa(i))
		if i%2 == 0 {
			channel.TxRollback()
		} else {
			channel.TxCommit()
		}
		timehelper.Sleep(1 * timehelper.Second)
		fmt.Println(i)
	}
	select {}
	//kutengone.RecieveRouting()
}

func test() {
	channel := giftRabbitMQ.OpenChannel()
	defer channel.Close()
	for {

		//
		bind, _ := giftRabbitMQ.RecieveRoutingWithOutBind(channel, false)
		for delivery := range bind {
			fmt.Println(string(delivery.Body))
			delivery.Ack(true)
		}
		timehelper.Sleep(2 * timehelper.Second)
	}
}

*/
